package com.hoshiicloud.elasticsearch.canal;

import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.hoshiicloud.elasticsearch.entity.Spu;
import com.hoshiicloud.elasticsearch.service.SpuService;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author ygd
 * @title: MyCommandLineRunner
 * @projectName elasticsearch
 * @description:
 * @date 2019/7/15
 */
@Component
@Order(10)
public class MyCommandLineRunner implements CommandLineRunner {

    private static Logger logger = LoggerFactory.getLogger(MyCommandLineRunner.class);

    @Value("${info.kafka.topic}")
    private String topic;

    @Value("${info.kafka.partition}")
    private Integer partition;

    @Value("${info.kafka.groupId}")
    private String groupId;

    @Value("${info.kafka.servers}")
    private String servers;

    @Autowired
    private SpuService spuService;

    private static boolean running = true;

    private static final List<String> tableNames = new ArrayList();

    {
        tableNames.add("ext_spu");
    }

    @Override
    public void run(String... args) throws Exception {
        System.out.println("监听器启动");
        canalkafkaConsumer();
    }

    public void canalkafkaConsumer() {
        final ExecutorService executor = Executors.newFixedThreadPool(10);
        final KafkaCanalConnector connector = new KafkaCanalConnector(servers, topic, partition, groupId, null, false);
        executor.submit(new Runnable() {

            @Override
            public void run() {
                try {
                    connector.connect();
                    connector.subscribe();
                    while (running) {
                        List<Message> messages = connector.getList(3L, TimeUnit.SECONDS);
                        if (messages != null) {
                            for (Message message : messages) {
                                long batchId = message.getId();
                                int size = message.getEntries().size();
                                if (batchId == -1 || size == 0) {
                                    // try {
                                    // Thread.sleep(1000);
                                    // } catch (InterruptedException e) {
                                    // }
                                } else {
                                    //printSummary(message, batchId, size);
                                    logger.info(message.toString());
                                    printEntry(message.getEntries());

                                }
                            }
                        }
                        connector.ack(); // 提交确认
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }

                connector.unsubscribe();
                connector.disconnect();
            }
        });

        try {
            Thread.sleep(50000);
        } catch (InterruptedException e) {
        }
        //running = false;
        //executor.shutdown();
        logger.info("shutdown completed");
    }

    protected void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;
            Date date = new Date(entry.getHeader().getExecuteTime());
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN) {
                    CanalEntry.TransactionBegin begin = null;
                    try {
                        begin = CanalEntry.TransactionBegin.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务头信息，执行的线程id，事务耗时
                    logger.info(entry.getHeader().getLogfileName(),
                            String.valueOf(entry.getHeader().getLogfileOffset()),
                            String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                            entry.getHeader().getGtid(), String.valueOf(delayTime));
                    logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
                    printXAInfo(begin.getPropsList());
                } else if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                    CanalEntry.TransactionEnd end = null;
                    try {
                        end = CanalEntry.TransactionEnd.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务提交信息，事务id
                    logger.info("----------------\n");
                    logger.info(" END ----> transaction id: {}", end.getTransactionId());
                    printXAInfo(end.getPropsList());
                    logger.info(entry.getHeader().getLogfileName(),
                            String.valueOf(entry.getHeader().getLogfileOffset()),
                            String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                            entry.getHeader().getGtid(), String.valueOf(delayTime));
                }

                continue;
            }

            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }

                CanalEntry.EventType eventType = rowChage.getEventType();

                logger.info(entry.getHeader().getLogfileName(),
                        String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                        entry.getHeader().getTableName(), eventType,
                        String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                        entry.getHeader().getGtid(), String.valueOf(delayTime));

                String tableName = entry.getHeader().getTableName();

                if (!tableNames.contains(tableName)) {
                    logger.info("跳过该表：" + tableName);
                    continue;
                }

                logger.info("表：" + tableName);

                if (eventType == CanalEntry.EventType.QUERY || rowChage.getIsDdl()) {
                    logger.info(" sql ----> " + rowChage.getSql());
                    continue;
                }

                printXAInfo(rowChage.getPropsList());
                for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                    if (eventType == CanalEntry.EventType.DELETE) {
                        delete(tableName, rowData.getBeforeColumnsList());
                    } else {
                        insertOrUpdate(tableName, rowData.getAfterColumnsList());
                    }
                }
            }
        }
    }

    protected void insertOrUpdate(String tableName, List<CanalEntry.Column> columns) {
        String id = null;
        for (CanalEntry.Column column : columns) {

            try {
                if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB")
                        || StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY")) {

                    if (!column.getName().equals("id")) {
                        continue;
                    }

                    id = new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8");

                } else {

                    if (!column.getName().equals("id")) {
                        continue;
                    }

                    id = column.getValue();
                }


            } catch (UnsupportedEncodingException e) {

            }

        }


        logger.info("id-------------------------" + id);

        Spu spu = spuService.create(Long.parseLong(id));

    }

    protected void delete(String tableName, List<CanalEntry.Column> columns) {
        String id = null;
        for (CanalEntry.Column column : columns) {
            StringBuilder builder = new StringBuilder();
            try {
                if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB")
                        || StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY")) {

                    if (!column.getName().equals("id")) {
                        continue;
                    }

                    id = new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8");

                } else {

                    if (!column.getName().equals("id")) {
                        continue;
                    }

                    id = column.getValue();
                }


            } catch (UnsupportedEncodingException e) {

            }
        }

        logger.info("id-------------------------" + id);

        spuService.delete(Long.parseLong(id));
    }

    protected static void printXAInfo(List<CanalEntry.Pair> pairs) {
        if (pairs == null) {
            return;
        }

        String xaType = null;
        String xaXid = null;
        for (CanalEntry.Pair pair : pairs) {
            String key = pair.getKey();
            if (StringUtils.endsWithIgnoreCase(key, "XA_TYPE")) {
                xaType = pair.getValue();
            } else if (StringUtils.endsWithIgnoreCase(key, "XA_XID")) {
                xaXid = pair.getValue();
            }
        }

        if (xaType != null && xaXid != null) {
            logger.info(" ------> xaType:" + xaType + "--- xaXid:" + xaXid);
        }
    }
}
